package com.ds.lens.data.spring.event;

import com.alibaba.fastjson.JSONObject;
import com.ds.lens.data.kafka.KafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

/**
 * Description:
 * Listen for AlarmRuleEvent，send message to kafka.
 *
 * @author WeiShaoying
 * @date 2019-07-02
 */
@Component
@Slf4j
public class AlarmRuleEventListener implements ApplicationListener<AlarmRuleEvent> {

    @Value("${lens.kafka.metadata.event.topic}")
    private String jobKafkaTopic;

    private final KafkaProducer kafkaProducer;

    @Autowired
    public AlarmRuleEventListener(KafkaProducer kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public void onApplicationEvent(AlarmRuleEvent event) {
        if (null != event.getEventTypeEnum() && null != event.getMetadataUpdateEvent()) {
            //Send message to Kafka
            kafkaProducer.sendMessage(jobKafkaTopic, event.getEventTypeEnum().name(), JSONObject.toJSONString(event.getMetadataUpdateEvent()));
        } else {
            log.error("AlarmRuleEventListener.onApplicationEvent error. AlarmRuleEvent:" + JSONObject.toJSONString(event));
        }
    }
}
